import com.alibaba.fastjson.JSON
import org.apache.log4j.{Level, Logger}
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.SQLContext
import org.apache.spark.{SparkConf, SparkContext}

/**
  * @Author: xuechunhua
  * @Date: 2020/6/2 15:14
  * @version 1.0
  */
object UpdateShopNamePriceXue {
  def main(args: Array[String]): Unit = {
    Logger.getLogger("org.apache.spark").setLevel(Level.WARN)
    val conf = new SparkConf()
    conf.setAppName(s"${this.getClass.getSimpleName}")
    conf.set("spark.debug.maxToStringFields", "500")
    conf.set("spark.testing.memory", "2147480000")
    conf.setMaster("local[*]")
    conf.set("es.nodes", "192.168.1.29")
    conf.set("es.port", "9200")
    conf.set("cluster.name", "O2OElastic")
    conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
    conf.set("spark.sql.caseSensitive", "true")
    val sc = new SparkContext(conf)
    val sqlContext = new SQLContext(sc)
    sc.setLogLevel("WARN")
    sc.hadoopConfiguration.set("fs.s3a.access.key", "GAO7EO9FWKPJ8WFCQDME")
    sc.hadoopConfiguration.set("fs.s3a.secret.key", "LZ0xaHBSYKHaJ9ECDbX9f7zin79UZkXfGoNapRPL")
    sc.hadoopConfiguration.set("fs.s3a.endpoint", "https://obs.cn-north-1.myhuaweicloud.com")
    /**
      * 天猫
      */
    val months = 10
//    val index = s"2021_tmall/tmall_2021_${months}"
//    val index = s"tmall_2021_activity/tmall_2021_618"
    val index = s"2021_taobao/taobao_2021_${months}"
    import org.elasticsearch.spark._
    /**
      *   79.00     618593263631
      */
    val sourceDate: RDD[String] = sc.esJsonRDD(s"${index}",
      """
        |  {
        |  "size": 10
        |  , "query": {
        |   "terms": {
        |     "shopName.keyword": [
        |       "雪梨国际美妆"
        |     ]
        |   }
        |  }
        |}
      """.stripMargin).values
    //  sqlContext.read.json(sourceDate).selectExpr("good_id").dropDuplicates("good_id").repartition(1).write.json("C:\\Users\\o2o-rd-0009\\Desktop\\华哥\\路之谐旗舰店")
    // sourceDate.repartition(1).saveAsTextFile("D:\\项目工程\\priceexp\\7月份异常价格\\616809140098\\")
    val sourcePriceEX = sourceDate.map(line => {
      val nObject = JSON.parseObject(line)
      var price = nObject.get("priceText").toString
      var new_price=0.0
      if(price.toDouble > 999){
//      if(price.toDouble == 0){
       //price = price/1000
        new_price = price.substring(0,2).toDouble
//        new_price = price.toDouble%200
//        new_price = 85
        //price = price -1000
       /* if(price >= 300){
          price = price -30
        }*/
      }else{
        new_price = price.toDouble

      }
      val sellCount = nObject.get("sellCount").toString().toInt
      nObject.put("priceText", new_price)
      nObject.put("salesAmount", (new_price.toDouble * sellCount).toDouble.formatted("%.2f"))
      nObject
    })
 //  sqlContext.read.json(sourcePriceEX).selectExpr("priceText").show(false)

     .saveToEs(s"${index}", Map("es.mapping.id" -> "good_id"))
    sc.stop()

  }
}
